package com.study.spring.kafka;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.*;


/**
 * @author Comsys-zhang
 * @ClassName: Consumer
 * @Description: 消费者
 * @date 2019年5月21日 下午10:07:24
 */
public class Consumer {

    private static final Logger LOG = LoggerFactory.getLogger(Consumer.class);

    private static final String BROKER_LIST = "192.168.1.4:9092";

    private static final String TOPIC = "visit";

    private static final KafkaConsumer<String, String> consumer;

    private static final ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(1, 1, 60L, TimeUnit.SECONDS, new SynchronousQueue<>());

    static {
        Properties configs = initConfig();
        consumer = new KafkaConsumer<>(configs);
        consumer.subscribe(Collections.singletonList(TOPIC));
    }

    private static Properties initConfig() {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKER_LIST);
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "zhangyi");
        properties.put("enable.auto.commit", "true");
        properties.put("auto.commit.interval.ms", "1000");
        properties.put("session.timeout.ms", "30000");
        properties.put("auto.offset.reset", "earliest");
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        return properties;
    }


    public static void main(String[] args) {

        Future<ConsumerRecords<String, String>> future = poolExecutor.submit(new Callable<ConsumerRecords<String, String>>() {
            @Override
            public ConsumerRecords<String, String> call() throws Exception {
                return consumer.poll(5);
            }
        });

        try {
            ConsumerRecords<String, String> records = future.get(10, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {

            e.printStackTrace();
        }
        poolExecutor.shutdown();
        consumer.close();

//		consumer.assignment();
//		consumer.close(10L,true);

    }
}
